// package com.zmn.oms.task.subscribe;
//
// import com.alibaba.fastjson.JSON;
// import com.aliyun.drc.client.message.ByteString;
// import com.aliyun.drc.client.message.DataMessage.Record;
// import com.aliyun.drc.client.message.DataMessage.Record.Field;
// import com.aliyun.drc.client.message.DataMessage.Record.Type;
// import com.aliyun.drc.clusterclient.ClusterListener;
// import com.aliyun.drc.clusterclient.message.ClusterMessage;
// import com.google.common.collect.Lists;
// import com.google.common.collect.Maps;
// import com.zmn.common.utils.string.StringUtil;
// import com.zmn.manager.mq.constant.MqTagConsts;
// import com.zmn.manager.mq.constant.MqTopicConsts;
// import com.zmn.manager.mq.interfaces.ZmnMQSender;
// import com.zmn.oms.common.constant.OrderConsts;
// import com.zmn.oms.common.constant.OrderStatusConsts;
// import com.zmn.oms.manager.utils.OmsMqTagConsts;
// import org.apache.commons.lang.StringUtils;
// import org.slf4j.Logger;
// import org.slf4j.LoggerFactory;
// import org.springframework.beans.factory.annotation.Autowired;
// import org.springframework.stereotype.Service;
//
// import java.io.UnsupportedEncodingException;
// import java.util.*;
//
// /**
//  * 类描述：
//  *
//  * @author lujia
//  * @date 2018/12/05 14:42
//  */
// @Service
// public class DtsOrderClusterListener extends ClusterListener {
//
//     private static final Logger logger = LoggerFactory.getLogger(DtsOrderClusterListener.class);
//
//     @Autowired
//     private ZmnMQSender zmnMQSender;
//
//     private static Map<String, String> tableMap = Maps.newHashMap();
//
//     /**
//      * 配件表
//      */
//     private static String PART_CHANGE_TABLE = "oms_order_part";
//
//     /**
//      * 评价表
//      */
//     private static String COMMENT_CHANGE_TABLE = "oms_order_comment_extend";
//
//     /**
//      * 工单表
//      */
//     private static String WORK_CHANGE_TABLE = "oms_order_work";
//
//     static {
//
//         // 这里维护订阅表与分片id映射
// //        tableMap.put("oms_order", "order_id");
//         tableMap.put("oms_order_work", "order_id");
//         tableMap.put("oms_order_detail", "detail_id");
//         tableMap.put("oms_order_member", "member_order_id");
//         tableMap.put("oms_order_work_master", "order_id");
//         tableMap.put("oms_order_product", "order_id");
//         tableMap.put("oms_order_service_item", "order_id");
//         tableMap.put("oms_order_discount", "order_id");
//         tableMap.put("oms_order_comment", "comment_id");
//         tableMap.put("oms_order_extend", "ext_id");
//         tableMap.put("oms_order_invoice", "invoice_id");
//         tableMap.put("oms_order_part", "order_id");
//         tableMap.put("oms_order_remark", "order_id");
//         tableMap.put("oms_order_comment_extend", "order_id");
//         tableMap.put("oms_order_factory", "fc_order_id");
//         tableMap.put("oms_order_factory_part", "order_id");
//         tableMap.put("oms_order_factory_part_post", "order_id");
//
//         tableMap.put("fico_clearing_master_order", "order_id");
//     }
//
//     @Override
//     public void notify(List<ClusterMessage> messages) throws Exception {
//         for (ClusterMessage message : messages) {
//             Record record = message.getRecord();
//             Type opt = record.getOpt();
//
//             // 只处理新增修改删除的操作
//             boolean edit = Objects.equals(Type.INSERT, opt) || Objects.equals(Type.UPDATE, opt) || Objects.equals(Type.DELETE, opt);
//             if (!edit) {
//                 message.ackAsConsumed();
//                 continue;
//             }
//
//             // 获取工单主键
//             Long orderId = getKey(record);
//             if (orderId == null) {
//                 message.ackAsConsumed();
//                 continue;
//             }
//
//             // 订单评价
//             if (Objects.equals(COMMENT_CHANGE_TABLE, record.getTablename()) || validateComment(record)) {
//                 sendOrderCommentMq(orderId);
//             }
//
//             // 配件 或者 算账 报销
//             if (record.getTablename().startsWith(PART_CHANGE_TABLE) || validatePart(record)) {
//                 sendOrderPartMq(orderId);
//             }
//
//             // 是否数据库批量变更中的第一条事务日志 否则跳过
//             if (!record.isFirstInLogevent()) {
//                 message.ackAsConsumed();
//                 continue;
//             }
//
//             // 同步工单
//             sendOrderMq(orderId);
//             sendOrderWorkMq(orderId);
//             // ackAsConsumed必须调用
//             message.ackAsConsumed();
//         }
//     }
//
//     /**
//      * 校验配件
//      * @param record
//      * @return
//      */
//     private boolean validatePart(Record record){
//         Type opt = record.getOpt();
//
//         if (!record.getTablename().startsWith(WORK_CHANGE_TABLE) || !Objects.equals(Type.UPDATE, opt)) {
//             return false;
//         }
//
//         List<Field> fieldList = record.getFieldList();
//         List<Field> newFieldList = Lists.newArrayListWithCapacity(fieldList.size());
//         for (int i = 0; i < fieldList.size(); i++) {
//             if (i % 2 == 0) {
//                 continue;
//             }
//             newFieldList.add(fieldList.get(i));
//         }
//
//         // 工单状态
//         Optional<Field> statusOptinal = newFieldList
//                 .stream()
//                 .filter((e) -> Objects.equals(e.getFieldname(), "status"))
//                 .findFirst();
//
//         // 算账时候需要同步配件
//         if (statusOptinal.isPresent()) {
//             ByteString value = statusOptinal.get().getValue();
//             try {
//                 Integer status = Integer.valueOf(value.toString("UTF-8"));
//                 if (Objects.equals(OrderStatusConsts.WORK_STATUS_ACCOUNT, status)) {
//                     return true;
//                 }
//             } catch (UnsupportedEncodingException e) {
//                 logger.error(e.getMessage(), e);
//                 return false;
//             }
//         }
//
//         // 配件报销状态
//         Optional<Field> partReimburseOptional= newFieldList
//                 .stream()
//                 .filter((e) -> Objects.equals(e.getFieldname(), "part_reimburse_status"))
//                 .findFirst();
//
//         // 配件报销需要同步配件
//         if (partReimburseOptional.isPresent()) {
//             ByteString value = partReimburseOptional.get().getValue();
//             try {
//                 Integer partReimburse = Integer.valueOf(value.toString("UTF-8"));
//                 if (Objects.equals(OrderConsts.PART_REIMBURSE_STATUS_FAIL, partReimburse)
//                 || Objects.equals(OrderConsts.PART_REIMBURSE_STATUS_SUCCESS, partReimburse)) {
//                     return true;
//                 }
//             } catch (UnsupportedEncodingException e) {
//                 logger.error(e.getMessage(), e);
//                 return false;
//             }
//         }
//
//         return false;
//     }
//
//     /**
//      * 校验评价
//      * @param record
//      * @return
//      */
//     private boolean validateComment(Record record){
//         Type opt = record.getOpt();
//
//         if (!record.getTablename().startsWith(WORK_CHANGE_TABLE) || !Objects.equals(Type.UPDATE, opt)) {
//             return false;
//         }
//
//         List<Field> fieldList = record.getFieldList();
//         List<Field> newFieldList = Lists.newArrayListWithCapacity(fieldList.size());
//         for (int i = 0; i < fieldList.size(); i++) {
//             if (i % 2 == 0) {
//                 continue;
//             }
//             newFieldList.add(fieldList.get(i));
//         }
//
//         // 工单回访人
//         Optional<Field> replyVisiterIdOptinal = newFieldList
//                 .stream()
//                 .filter((e) -> Objects.equals(e.getFieldname(), "reply_visiter_id"))
//                 .findFirst();
//
//         // 算账时候需要同步配件
//         if (replyVisiterIdOptinal.isPresent() && replyVisiterIdOptinal.get().getValue() != null) {
//             ByteString value = replyVisiterIdOptinal.get().getValue();
//             try {
//                 Long replyVisiterId = Long.valueOf(value.toString("UTF-8"));
//                 if (replyVisiterId != null) {
//                     return true;
//                 }
//             } catch (UnsupportedEncodingException e) {
//                 logger.error(e.getMessage(), e);
//                 return false;
//             }
//         }
//
//         return false;
//     }
//
//     /**
//      * 获取Id
//      * @param record
//      * @return
//      */
//     private Long getKey(Record record){
//         String tableName = record.getTablename();
//         String key = tableMap.get(tableName);
//
//         if (StringUtils.isEmpty(key)) {
//             String simpleTableName = tableName.substring(0, tableName.length() - 2);
//             key = tableMap.get(simpleTableName);
//         }
//
//         if (StringUtil.isBlank(key)) {
//             return null;
//         }
//
//         String finalKey = key;
//
//         // 找到主键值 同步工单
//         Optional<Field> first = record.getFieldList()
//                 .stream()
//                 .filter((e) -> Objects.equals(e.getFieldname(), finalKey))
//                 .findFirst();
//
//         if (first.isPresent()) {
//             ByteString value = first.get().getValue();
//             Long orderId = null;
//             try {
//                 orderId = Long.valueOf(value.toString("UTF-8"));
//             } catch (UnsupportedEncodingException e) {
//                 logger.error(e.getMessage(), e);
//             }
//             return orderId;
//         }
//         return null;
//     }
//
//     @Override
//     public void noException(Exception e) {
//         logger.error(e.getMessage(), e);
//     }
//
//     /**
//      * 工单
//      * @param orderId
//      */
//     private void sendOrderWorkMq(Long orderId){
//         String key = StringUtil.getRandomsString(32);
//         logger.debug("#oms#dts#orderwork#sendMq key : {} orderId : {}", key, orderId);
//
//         List<Long> orderIds = Arrays.asList(orderId);
//         zmnMQSender.send(MqTopicConsts.ZMN_TOPIC_OMS, MqTagConsts.ZMN_TOPIC_OMS_CHANGE_ORDER_WORK, key, JSON.toJSONString(orderIds));
//     }
//
//     /**
//      * 订单
//      * @param orderId
//      */
//     private void sendOrderMq(Long orderId){
//         String key = StringUtil.getRandomsString(32);
//         logger.debug("#oms#dts#order#sendMq key : {} orderId : {}", key, orderId);
//
//         zmnMQSender.send(MqTopicConsts.ZMN_TOPIC_OMS, MqTagConsts.ZMN_TOPIC_OMS_CHANGE_ORDER, key, String.valueOf(orderId));
//     }
//
//     /**
//      * 订单配件
//      * @param orderId
//      */
//     private void sendOrderPartMq(Long orderId){
//         String key = StringUtil.getRandomsString(32);
//         logger.debug("#oms#dts#orderpart#sendMq key : {} orderId : {}", key, orderId);
//
//         List<Long> orderIds = Arrays.asList(orderId);
//         zmnMQSender.send(MqTopicConsts.ZMN_TOPIC_OMS, MqTagConsts.ZMN_TOPIC_OMS_CHANGE_ORDER_PART, key, JSON.toJSONString(orderIds));
//     }
//
//     /**
//      * 订单评价
//      * @param orderId
//      */
//     private void sendOrderCommentMq(Long orderId){
//         String key = StringUtil.getRandomsString(32);
//         logger.debug("#oms#dts#ordercomment#sendMq key : {} orderId : {}", key, orderId);
//         List<Long> orderIds = Arrays.asList(orderId);
//         zmnMQSender.send(MqTopicConsts.ZMN_TOPIC_OMS, OmsMqTagConsts.ZMN_TOPIC_OMS_CHANGE_ORDER_COMMENT, key, JSON.toJSONString(orderIds));
//     }
//
// }
